package com.ds.lens.flink.monitor.datasource.function;

import com.ds.lens.data.common.document.datasource.LensDatasource;
import com.ds.lens.flink.monitor.metricsjob.Metric;
import com.ds.lens.flink.monitor.metricsjob.Metrics;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author WeiShaoying
 * @date 2020/4/28 下午2:07
 */
public class DataSourceProcess extends ProcessFunction<Row, Metrics> {
    private LensDatasource dataSource;
    private Map<String, Integer> metricIndex;
    private static final long time_difference = 28800000;

    public DataSourceProcess(LensDatasource dataSource) {
        this.dataSource = dataSource;
        metricIndex = new HashMap<>(dataSource.getMetric().size());
        String[] fields = dataSource.getDsSql().getSelect().split(",");
        for (int i = 0; i < fields.length; i++) {
            if (fields[i].contains("as")) {
                metricIndex.put(fields[i].split("as")[1].trim(), i);
            } else {
                metricIndex.put(fields[i].trim(), i);
            }
        }
    }

    @Override
    public void processElement(Row row, Context context, Collector<Metrics> collector) throws Exception {
        Metrics metrics = new Metrics();
        metrics.setAppId(dataSource.getApplicationId());
        metrics.setAppKey(dataSource.getApplicationKey());
        List<Metric> metricList = new ArrayList<>();
        Map<String, String> tags = new HashMap<>();
        long timestamp = Timestamp.valueOf(row.getField(row.getArity() - 1).toString()).getTime() + time_difference;
        dataSource.getTags().forEach((k, v) -> {
            tags.put(getValue(row, k), getValue(row, v));
        });
        dataSource.getMetric().forEach((k, v) -> {
            metricList.add(createMetric(getValue(row, k), Double.valueOf(getValue(row, v)), timestamp, tags));
        });
        metrics.setMetrics(metricList);
        collector.collect(metrics);
    }

    private String getValue(Row row, String value) {
        if (value.contains("#")) {
            Object field = row.getField(metricIndex.get(value.replace("#", "")));
            return field == null ? null : field.toString();
        }
        return value;
    }

    private Metric createMetric(String key, double value, long time, Map<String, String> tags) {
        Metric metric = new Metric();
        metric.setMetricName(key);
        metric.setTags(tags);
        List<List<Double>> points = new ArrayList<>();
        List<Double> point = new ArrayList<>();
        point.add((double) time);
        point.add(value);
        points.add(point);
        metric.setPoints(points);
        return metric;
    }
}
